package com.hq.tomato.ms.aop;

import com.rabbitmq.client.Channel;
import com.hq.tomato.ms.annotation.ReliabilityStrategy;
import com.hq.tomato.ms.excpetion.MessageRepeatException;
import com.hq.tomato.ms.strategy.RequeueDlqStrategy;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *
 *
 * @Description: <br> 消息幂等切面
 * @Project: <br>
 * @CreateDate: Created in 2019-11-19 16:16 <br>
 * @Author: <a href="1026365768@qq.com">qiang.hu</a>
 */

@Aspect
@Slf4j
public class IdempotentMessageAspect {

    private final String IDEMPOTENT_PRE = "IDEMPOTENT:";

    /**消息过期时间保存5分钟**/
    private final long EXPIRE_TIME = 5;

    /**防消息在最后提交是发生网络抖动时间5S**/
    private final long EXPIRE_TIME_END = 5;


    @Autowired
    private RedisTemplate<String,String> redisTemplate;



    @Around(value = "@annotation(reliabilityStrategy)")
    public Object process(ProceedingJoinPoint joinPoint, ReliabilityStrategy reliabilityStrategy) throws Throwable {
        //幂等性操作
        String msgId = null;
        try {
            msgId = doIdempotent(joinPoint,reliabilityStrategy);
            log.info("msgId:{}",msgId);
        }catch (MessageRepeatException mee){
           //消息重复 不执行
            log.warn(mee.getMessage());
           return null;
        } catch (Exception e){
            log.warn("【beforeConsumer异常,异常信息为:{}】",e.getMessage());
        }


        //幂等性操作异常不能操作消息正常消费
        try {
            return joinPoint.proceed();
        }catch (Exception e){

            boolean needReturn = processRequeue(joinPoint,reliabilityStrategy);
            if(needReturn){
                return null;
            }
            log.warn("【消费执行消息异常,异常消息为:{}】",e.getMessage());
            throw e;
        }finally {
            if(Objects.nonNull(msgId)){
                //防止消息在5S内未能及时返回ack 而此时发生网络抖动的情况 所以此处选择不删除 而是将过期时间设置为5S后自动删除
                redisTemplate.expire(IDEMPOTENT_PRE+msgId,EXPIRE_TIME_END,TimeUnit.SECONDS);
            }
        }

    }

    private boolean processRequeue(ProceedingJoinPoint joinPoint, ReliabilityStrategy reliabilityStrategy) {
        //如果是重入队列策略 并且配置了Channel 则拒绝消息 并且重入队列
        if(!reliabilityStrategy.strategy().equals(RequeueDlqStrategy.class)){
            return false;
        }

        Object[] args = joinPoint.getArgs();
        if(Objects.isNull(args)||args.length==0) {
            log.warn("【策略为重入队列,但没有任何的参数!!!】");
            return true;
        }


        Message message = getMessageByArgs(args);

        if(Objects.isNull(message)){
            log.warn("【策略为重入队列,但未找到message!!!】");
            return true;
        }

        Channel channel = getChannelByArgs(args);
        if(Objects.isNull(channel)){
            log.warn("【策略为重入队列,但未找到channel,消息信息为:】");
            return true;
        }


        Object deliveryTag = message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicReject(Long.valueOf(deliveryTag.toString()),true);
        }catch (Exception e){
            log.warn("【策略为重入队列,拒绝消息异常,异常信息为:{}】",e.getMessage());
            return true;
        }

        return true;

    }


    private String doIdempotent(ProceedingJoinPoint joinPoint, ReliabilityStrategy reliabilityStrategy) {
        String msgId = null;
        boolean idempotent = reliabilityStrategy.idempotent();
        //幂等处理
        Object[] args = joinPoint.getArgs();
        if(args!=null&&args.length>0){
            Message message = getMessageByArgs(args);

            if(Objects.isNull(message)){
                return null;
            }

            //消息幂等
            if(idempotent){

                ValueOperations<String, String> opsForValue = redisTemplate.opsForValue();
                MessageHeaders headers = message.getHeaders();

                if(Objects.isNull(headers)){
                    return null;
                }

                //标识重试次数
                AtomicInteger deliveryAttempt = (AtomicInteger) headers.get("deliveryAttempt");

                //重试不校验幂等
                if(deliveryAttempt.get()>1){
                    return null;
                }

                //TODO 将stream自带的messageId 换成AMQP messageId
                Object id = headers.get(AmqpHeaders.MESSAGE_ID);

                if(Objects.isNull(id)){
                    return null;
                }

                msgId = id.toString();

                Boolean setSuccess = opsForValue.setIfAbsent(msgId, "");
                //如果消息已经在消费了 则不再重复消息
                if(!setSuccess){
                    throw new MessageRepeatException("消息重复,重复id为:"+msgId+",body为:"+message.getPayload());
                }
                //5分钟内防止消息幂等-> 一条消息在5分钟内必须执行完
                //幂等发生的情况 网络抖动 -> 当消息执行到一半时 发生网路抖动 此时mq 将会重新推送
                redisTemplate.expire(IDEMPOTENT_PRE + msgId, EXPIRE_TIME, TimeUnit.MINUTES);
            }
        }

        return msgId;
    }



    /**
     * @Author huqiang
     * @Description //TODO 通过参数 获取Message对象
     * @Date  2019-11-19
     * @Param args
     * @return org.springframework.messaging.Message
     */
    private Message getMessageByArgs(Object [] args){
        Message message = Arrays.stream(args)
                .filter(obj -> obj instanceof Message)
                .map(obj -> (Message) obj)
                .findFirst()
                .orElse(null);
        return message;
    }


    /**
     * @Author huqiang
     * @Description //TODO 通过参数 获取Message对象
     * @Date  2019-11-19
     * @Param args
     * @return org.springframework.messaging.Message
     */
    private Channel getChannelByArgs(Object [] args){
        Channel channel = Arrays.stream(args)
                .filter(obj -> obj instanceof Channel)
                .map(obj -> (Channel) obj)
                .findFirst()
                .orElse(null);
        return channel;
    }


}
